package com.ah._05oneway;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消费者
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        //创建消费者对象
        DefaultMQPushConsumer dmpc = new DefaultMQPushConsumer("oneway_consumer_group");
        dmpc.setNamesrvAddr("192.168.19.132:9876");
        //设置订阅信息
        dmpc.subscribe("oneway_topic","*");
        //监听topic信息
        dmpc.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt: list ){
                    byte[] body = messageExt.getBody();
                    String msg = new String(body);
                    System.out.println("msg = " + msg);
                }
                //返回消费状态
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //启动消费者
        dmpc.start();

        //永远运行下去
        Thread.sleep(Long.MAX_VALUE);

        //关闭消费者
//        dmpc.shutdown();

    }
}
